/*
 * Copyright (c)  2020,  DataPipeline
 * All rights reserved.
 *
 * Id:WordCountApplication.java   2020-07-26 21:18 wanglijun
 */
package cn.datacare.kafka.stream.wordcount;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;
import java.util.Properties;

/**
 * <p>
 * Title:
 * </p>
 * <p>
 * Description:
 * </p>
 * <p>
 * Copyright: Copyright (c) 2020
 * </p>
 * <p>
 * Company/Department: DataPipeline
 * </p>
 *
 * @author wanglijun
 * @version 1.0
 * @since 2020-07-26 21:18
 */
public class WordCountApplication {

    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("TextLinesTopic2");
        KTable<String, Long> wordCounts = textLines
                // Create new KStream: by implements {@link ValueMapper} transform map textLine to String array
                .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
                // select value as key
                .selectKey((key, word) -> word)
                .groupByKey()
                // count the number of records in this new stream. use default store
                .count(Materialized.as("counts-store"));
        // write this new kstream into another Kafka topic named WordsWithCountsTopic
        wordCounts.toStream().to("WordsWithCountsTopic2", Produced.with(Serdes.String(), Serdes.Long()));

        // inspect what kinds of topology is created
        Topology topology = builder.build();
        System.out.println(topology.describe());

        // create kafka stream
        KafkaStreams streams = new KafkaStreams(topology, getProps());
        streams.start();

    }

    /**
     * 配置文件
     * @return Properties
     */
    private static Properties getProps(){
        Properties props=new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"wordCount-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return props;
    }
}
